package net.bwie.realtime.vehicle.dwd.job.one;


import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import net.bwie.realtime.utils.DateTimeUtil;
import net.bwie.realtime.utils.DorisUtil;
import net.bwie.realtime.utils.KafkaUtil;
import net.bwie.realtime.vehicle.dwd.bean.MonitorInfo;
import net.bwie.realtime.vehicle.dwd.function.OverSpeedRichFilterFunction;
import net.bwie.realtime.vehicle.dwd.utils.OverSpeedUtil;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;

/*
    TODO 实时车辆超速监控
     通过实时计算，把所有超速超过10%的车辆找出来，写入关系型数据库中
        没有限速的的路口，默认限速60
 */
/**
 * @author ASUS
 * 实时车辆超速监控
 */

public class OverSpeedController {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.enableCheckpointing(5000L);
        // 获取数据
        DataStream<String> dwdCarMonitor = KafkaUtil.consumerKafka(env, "dwd_car_monitor");

//        dwdCarMonitor.print();
        // 处理数据 拉取限速
        DataStream<String> ds1 = processJson(dwdCarMonitor);
//        ds1.print();
        // 过滤超速
        DataStream<String> stringDataStream = monitorFilter(ds1);

//        stringDataStream.print();

        // 转换数据结构
        DataStream<String> monitor = stringDataStream.map(new MapFunction<String, String>() {

            @Override
            public String map(String s) throws Exception {
                MonitorInfo monitorInfo = JSON.parseObject(s, MonitorInfo.class);
                String car = monitorInfo.getCar();
                String monitorId = monitorInfo.getMonitorId();
                String roadId = monitorInfo.getRoadId();
                Integer speedLimit = monitorInfo.getSpeedLimit();
                Integer speed = monitorInfo.getSpeed();
                String s1 = DateTimeUtil.convertLongToString(monitorInfo.getActionTime(), "yyyy-MM-dd HH:mm:ss");
                return String.format(
                        "{\"car\":\"%s\",\"monitorId\":\"%s\",\"roadId\":\"%s\",\"realSpeed\":\"%d\",\"limitSpeed\":\"%d\",\"actionTime\":\"%s\"}",
                        car,
                        monitorId,
                        roadId,
                        speed,
                        speedLimit,
                        s1);
            }
        });

        // 打印结果
        monitor.print();


        // 数据写入Kafka
        KafkaUtil.producerKafka(stringDataStream,"dwd_speeding_info");
        // 数据写入Doris
        DorisUtil.saveToDoris(monitor,"vehicle_realtime_report","t_speeding_info");


        env.execute("OverSpeedController");
    }


    private static DataStream<String> monitorFilter(DataStream<String> ds1) {
        return ds1.filter(new OverSpeedRichFilterFunction<>());
    }


    private static DataStream<String> processJson(DataStream<String> dwdCarMonitor) {

        return dwdCarMonitor.map(new MapFunction<String, String>() {
            @Override
            public String map(String s) throws Exception {
                JSONObject jsonObject = JSON.parseObject(s);
                String monitorId = jsonObject.getString("monitorId");

                // 从MonitorUtil的缓存中获取限速信息，而不是重新加载
                // 注意：需要在MonitorUtil中添加一个获取缓存的方法
                Integer speedLimit = OverSpeedUtil.getSpeedLimitFromCache(monitorId);


                // 如果缓存中没有，使用默认值（例如60）
                if (speedLimit == null) {
                    speedLimit = 60;
                    // 可选：如果需要，可以在这里触发一次数据库查询加载该monitorId的信息
                     OverSpeedUtil.preloadAreaInfo(monitorId);
                }

                // 将获取到的限速信息放入JSON对象
                jsonObject.put("speedLimit", speedLimit);

                return jsonObject.toJSONString();
            }
        });
    }
}
